package com.dayu.finecomm.iot.service;

import static io.netty.handler.codec.http.HttpHeaders.Names.CONTENT_TYPE;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.http.DefaultFullHttpRequest;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpVersion;

import java.net.URI;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.dayu.finecomm.iot.IotHttpClientUtil;
import com.dayu.finecomm.mongo.pojo.IotPlat;
import com.dayu.finecomm.mongo.service.IotPlatService;

/**
 * IotHistoryDataDamon, 处理历史数据上报
 * 
 * @author XuJialiang
 * @since 2017-11-01 10:22:12
 * @version V1.0.0
 */
public class IotHistoryDataDamon implements Runnable {
	static Logger LOGGER = LoggerFactory.getLogger(IotHistoryDataDamon.class);

	/**
	 * IOT Plat 数据库
	 */
	private IotPlatService mMongoDB;

	/**
	 * unitalk回调地址
	 */
	private String mRemoteHost;

	/**
	 * unitalk回调地址端口
	 */
	private int mRemotePort;

	/**
	 * 存放每次请求的历史数据
	 */
	LinkedList<JSONObject> mHistoryDataList;

	/**
	 * 锁
	 */
	Lock mLock;

	public IotHistoryDataDamon(IotPlatService mongoDB, String host, int port) {
		mMongoDB = mongoDB;

		if (null == host || host.isEmpty()) {
			mRemoteHost = new String("127.0.0.1");
		} else {
			mRemoteHost = host;
		}

		mRemotePort = port;

		mHistoryDataList = new LinkedList<JSONObject>();

		mLock = new ReentrantLock();

		ScheduledExecutorService service = Executors
				.newSingleThreadScheduledExecutor();
		service.scheduleAtFixedRate(this, 10, 60, TimeUnit.SECONDS); // 每隔30秒执行一次
	}

	/**
	 * 插入历史数据
	 * 
	 * @param jsonObj
	 * @return
	 */
	public int putHistoryData(JSONObject jsonObj) {
		int result = -1;

		try {
			mLock.lock();

			if (0 == mHistoryDataList.size()) {
				mHistoryDataList.add(jsonObj);

				result = 0;
			}
		} finally {
			mLock.unlock();
		}

		return result;
	}

	public void run() {
		JSONObject jsonRoot = null;

		try {
			mLock.lock();

			jsonRoot = mHistoryDataList.pollLast();
		} finally {
			mLock.unlock();
		}

		if (null != jsonRoot) {
			try {
				Map<String, LinkedList<JSONObject>> datasMap = new HashMap<String, LinkedList<JSONObject>>();

				JSONObject jsonObj = jsonRoot.getJSONObject("body");
				JSONArray jsonDatas = jsonObj
						.getJSONArray("deviceDataHistoryDTOs");

				for (int i = 0; i < jsonDatas.size(); i++) {
					JSONObject jsonItem = jsonDatas.getJSONObject(i);

					String timestamp = jsonItem.getString("timestamp");

					if (datasMap.containsKey(timestamp)) {
						LinkedList<JSONObject> list = datasMap.get(timestamp);
						list.add(jsonItem);
					} else {
						LinkedList<JSONObject> list = new LinkedList<JSONObject>();
						list.add(jsonItem);
						datasMap.put(timestamp, list);
					}
				}

				String appid = jsonObj.getString("appId");
				IotPlat iotPlat = mMongoDB.query(appid);
				String protocol = iotPlat.getProtocol();

				for (Entry<String, LinkedList<JSONObject>> entry : datasMap
						.entrySet()) {
					LinkedList<JSONObject> list = entry.getValue();

					String jsonStr = packIotJson(protocol, list);
					if (null != jsonStr) {
						sendIotJson(appid, jsonStr);

						Thread.sleep(50);
					}
				}
			} catch (Exception e) {
				LOGGER.error("IotHistoryDataDamon run error: " + e);

			} finally {
				LOGGER.info("IotHistoryDataDamon send iot data finished!!!");
			}
		}
	}

	/**
	 * 组装Iot Json报文
	 * 
	 * @param protocol
	 * @param list
	 * @return
	 */
	private String packIotJson(String protocol, LinkedList<JSONObject> list) {
		String jsonString = null;

		try {
			if (0 == protocol.compareTo("YL809")) {
				if (null != list && 4 == list.size()) {
					JSONObject jsonObj = new JSONObject();

					JSONObject item_1 = list.get(0);
					JSONObject item_2 = list.get(1);
					JSONObject item_3 = list.get(2);
					JSONObject item_4 = list.get(3);

					JSONObject data_1 = new JSONObject();
					data_1.put("serviceId", item_1.get("serviceId"));
					data_1.put("serviceType", item_1.get("serviceId"));
					data_1.put("data", item_1.getJSONObject("data"));
					data_1.put("eventTime", item_1.get("timestamp"));

					JSONObject data_2 = new JSONObject();
					data_2.put("serviceId", item_2.get("serviceId"));
					data_2.put("serviceType", item_2.get("serviceId"));
					data_2.put("data", item_2.getJSONObject("data"));
					data_2.put("eventTime", item_2.get("timestamp"));

					JSONObject data_3 = new JSONObject();
					data_3.put("serviceId", item_3.get("serviceId"));
					data_3.put("serviceType", item_3.get("serviceId"));
					data_3.put("data", item_3.getJSONObject("data"));
					data_3.put("eventTime", item_3.get("timestamp"));

					JSONObject data_4 = new JSONObject();
					data_4.put("serviceId", item_4.get("serviceId"));
					data_4.put("serviceType", item_4.get("serviceId"));
					data_4.put("data", item_4.getJSONObject("data"));
					data_4.put("eventTime", item_4.get("timestamp"));

					jsonObj.put("notifyType", "deviceDatasChanged");
					jsonObj.put("requestId", "historyDeviceDatas");
					jsonObj.put("deviceId", item_1.get("deviceId"));
					jsonObj.put("gatewayId", item_1.get("gatewayId"));

					JSONArray jsonArray = new JSONArray();
					jsonArray.add(data_1);
					jsonArray.add(data_2);
					jsonArray.add(data_3);
					jsonArray.add(data_4);

					jsonObj.put("services", jsonArray);

					jsonString = jsonObj.toString();
				} else {
					LOGGER.error("YL809 packIotJson list is null or illegal");
				}
			} else if (0 == protocol.compareTo("HD86")) {
				if (null != list && 2 == list.size()) {
					JSONObject jsonObj = new JSONObject();

					JSONObject item_1 = list.get(0);
					JSONObject item_2 = list.get(1);

					JSONObject data_1 = new JSONObject();
					data_1.put("serviceId", item_1.get("serviceId"));
					data_1.put("serviceType", item_1.get("serviceId"));
					data_1.put("data", item_1.getJSONObject("data"));
					data_1.put("eventTime", item_1.get("timestamp"));

					JSONObject data_2 = new JSONObject();
					data_2.put("serviceId", item_2.get("serviceId"));
					data_2.put("serviceType", item_2.get("serviceId"));
					data_2.put("data", item_2.getJSONObject("data"));
					data_2.put("eventTime", item_2.get("timestamp"));

					jsonObj.put("notifyType", "deviceDatasChanged");
					jsonObj.put("requestId", "historyDeviceDatas");
					jsonObj.put("deviceId", item_1.get("deviceId"));
					jsonObj.put("gatewayId", item_1.get("gatewayId"));

					JSONArray jsonArray = new JSONArray();
					jsonArray.add(data_1);
					jsonArray.add(data_2);

					jsonObj.put("services", jsonArray);

					jsonString = jsonObj.toString();
				} else {
					LOGGER.error("HD86 packIotJson list is null or illegal");
				}
			} else {
				LOGGER.error("Protocol " + protocol + "is not a iot protocol");
			}
		} catch (Exception e) {
			LOGGER.error("packIotJson error with protocol " + protocol);

			jsonString = null;
		}

		return jsonString;
	}

	/**
	 * 发送报文给unitalk 回调接口
	 * 
	 * @param appid
	 * @param json
	 */
	private void sendIotJson(String appid, String json) throws Exception {
		IotHttpClientUtil client = new IotHttpClientUtil();

		URI uri = new URI("http://" + mRemoteHost + ":" + mRemotePort
				+ "/ycb/iot/" + appid + "/deviceDatasChanged");

		DefaultFullHttpRequest request = new DefaultFullHttpRequest(
				HttpVersion.HTTP_1_1, HttpMethod.POST, uri.getRawPath(),
				Unpooled.wrappedBuffer(json.getBytes()));

		request.headers().set(CONTENT_TYPE, "application/json");
		request.headers().set(HttpHeaders.Names.CONNECTION,
				HttpHeaders.Values.KEEP_ALIVE);
		request.headers().set(HttpHeaders.Names.CONTENT_LENGTH,
				request.content().readableBytes());

		JSONObject jsonObj = client.doHttpPost(uri.toString(), json,
				IotHttpClientUtil.Method.POST, "application/json");
		int code = jsonObj.getIntValue("code");
		if (200 != code) {
			LOGGER.error("sendIotJson ERROR: " + code);
		}
	}
}
